package org.zookeeper.examples.command.overview.component;

import java.util.Arrays;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/**
 * 负责监视znode节点变化并做出响应
 *
 * @author slydm
 */
public class DataMonitor implements Watcher, StatCallback {

  private ZooKeeper zooKeeper;
  private String znode;
  private boolean dead = false;
  private DataMonitor.MonitorLinstener monitorLinstener;
  private byte[] prevData;

  public DataMonitor(ZooKeeper zooKeeper, String znode,
      DataMonitor.MonitorLinstener monitorLinstener) {
    this.zooKeeper = zooKeeper;
    this.znode = znode;
    this.monitorLinstener = monitorLinstener;

    // 在节点注册watch
    zooKeeper.exists(znode, true, this, null);
  }
  public boolean isDead() {
    return dead;
  }

  public void setDead(boolean dead) {
    this.dead = dead;
  }

  /**
   * 被监视znode变化时被回调
   *
   * @param event
   */
  @Override
  public void process(WatchedEvent event) {
    System.out.println("DataMonitor : 监视到节点发生变化...." + event);

    String path = event.getPath();
    if (EventType.None == event.getType()) {

      switch (event.getState()) {
        case SyncConnected:
          break;
        case Expired:
          setDead(true);
          monitorLinstener.close(Code.SESSIONEXPIRED.intValue());
          break;
      }

    } else {

      if (path != null && path.equals(znode)) {
        zooKeeper.exists(path, true, this, null);
      }

    }
  }

  @Override
  public void processResult(int rc, String path, Object ctx, Stat stat) {
    // 判断返回状态, 确定服务器上是否存在该znode.
    boolean exists;
    switch (Code.get(rc)) {
      case OK:
        exists = true;
        break;
      case NONODE:
        exists = false;
        break;
      case SESSIONEXPIRED:
      case NOAUTH:
        dead = true;
        monitorLinstener.close(rc);
        return;
      default:
        // 重试
        zooKeeper.exists(znode, true, this, null);
        return;
    }

    // 存在znode,读取节点关联的数据
    byte[] b = null;
    if (exists) {

      try {
        b = zooKeeper.getData(znode, false, null);
      } catch (KeeperException e) {
        e.printStackTrace();
      } catch (InterruptedException e) {
        e.printStackTrace();
        return;
      }

      boolean isNodeDataModified =
          (b != null && b != prevData) || b != null && !Arrays.equals(b, prevData);
      if (isNodeDataModified) {
        monitorLinstener.exists(b);
        prevData = b;
      }

    }

  }

  /**
   * 连接状态监听器
   */
  public interface MonitorLinstener {

    /**
     * 被监听节点数据变化时该方法被回调
     *
     * @param data
     */
    void exists(byte[] data);

    /**
     * 客户端与zookeeper服务端连接过期时触发
     *
     * @param rc
     */
    void close(int rc);

  }

  public abstract class AbstractMonitorLinstener implements MonitorLinstener {

    @Override
    public void close(int rc) {

    }
  }
}
